package com.kafka.kafka;

import com.kafka.entity.Message;
import com.kafka.logs.SystemLogger;
import com.kafka.utils.ProtoStuffSerializeUtils;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.stereotype.Component;

/**
 * @ClassName KafaSender
 * @Description TODD
 * @Author Jesson
 * @Date 2019/1/14 15:46
 * @Version 1.0
 **/
@Component
public class KafkaProducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;
    @Autowired
    private KafkaProducerConfig kafkaProducerConfig;

    public void send(Message msg) {

        // Serialize
        byte[] message = ProtoStuffSerializeUtils.serialize(msg, Message.class);
        if (message == null) {
            return;
        }
        //发送消息
        kafkaTemplate.send(kafkaProducerConfig.getTopic(), message);

        //异步监听回调,实现对消息发送结果的回调，成功-onSuccess，失败-onError
        kafkaTemplate.setProducerListener(new ProducerListener<String, byte[]>() {
            @Override
            public void onSuccess(String topic, Integer partition, String key, byte[] value, RecordMetadata recordMetadata) {
            }

            @Override
            public void onError(String topic, Integer partition, String key, byte[] value, Exception exception) {
                SystemLogger.error("send error! topic:{}, partition:{}, key:{}, exception:{}",
                        topic, partition, key, exception.getMessage());
            }
        });
    }

}
